package main

import (
	"context"
	"github.com/Shopify/sarama"
	"github.com/astaxie/beego/logs"
	"strings"
)

type KafkaClient struct {
	consumer      sarama.ConsumerGroup
}

func NewKafkaClient() *KafkaClient {
	return &KafkaClient{
		consumer:      kafkaConsumer,
	}
}

var kafkaConsumer sarama.ConsumerGroup

type Handler struct {
	message func(message *sarama.ConsumerMessage)
}

func (h *Handler) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
// but before the offsets are committed for the very last time.
func (h *Handler) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
// Once the Messages() channel is closed, the Handler must finish its processing
// loop and exit.
func (h *Handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		h.message(message)
		session.MarkMessage(message, "")
	}
	return nil
}

func init() {
	brokerListStr := "192.168.3.14:19092,192.168.3.50:19092,192.168.3.59:19092"
	logs.Info("Init kafka addr:%s", brokerListStr)
	brokerList := strings.Split(brokerListStr, ",")
	config := sarama.NewConfig()
	config.Version = sarama.MaxVersion

	//consumerConfig.Consumer.Offsets.CommitInterval = time.Second
	groupId := "rtls-notice-test"
	consumer, err := sarama.NewConsumerGroup(brokerList, groupId, config)
	if err != nil {
		logs.Error("kafka connect error:%v", err.Error())
		return
	}
	kafkaConsumer = consumer
}

func (kc *KafkaClient) RecvMsg(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
	for {
		select {
		case <-ctx.Done():
			logs.Warn("terminating: context cancelled")
			return
		default:
			if err := kc.consumer.Consume(ctx, topics, handler); err != nil {
				logs.Error("Error from consumer: %v", err)
			}
		}
	}
}

func main() {
	NewKafkaClient().RecvMsg(context.Background(), []string{"TagPosition"}, &Handler{message: func(message *sarama.ConsumerMessage) {
		logs.Info(string(message.Value))
	}})
}
